package com.kevin.aqmp.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Map;

/**
 * 生产者发送到Broker中接口Confirm
 *
 * @author Kevin Liu
 * @date 2020/8/11 下午8:44
 */
@Component
@Slf4j
public class RabbitmqConfirmCallback  implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        if (b) {
            log.info("消息投递成功~消息Id：{}", correlationData.getId());
        } else {
            log.error("消息投递失败，Id：{}，错误提示：{}", correlationData.getId(), s);
        }
    }

    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {
        log.info("消息没有路由到队列，获得返回的消息");
        Map map = byteToObject(message.getBody(), Map.class);
        log.info("message body: {}", map == null ? "" : map.toString());
        log.info("replyCode: {}", i);
        log.info("replyText: {}", s);
        log.info("exchange: {}", s1);
        log.info("routingKey: {}", s2);
        log.info("------------> end <------------");
    }

    @SuppressWarnings("unchecked")
    private <T> T byteToObject(byte[] bytes, Class<T> clazz) {
        T t;
        try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
             ObjectInputStream ois = new ObjectInputStream(bis)) {
            t = (T) ois.readObject();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
        return t;
    }
}
